package io.openmessaging;

import io.openmessaging.io.FileChannelWrapper;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 产生topicName 和 topicId 的对应关系，并持久化到硬盘上
 *
 * @author cuijd    2021/9/26 14:56
 */
public class PersistentTopicMap {

    private ConcurrentHashMap<String, Topic> topics = new ConcurrentHashMap<>(Conf.maxTopicId);
    private FileChannelWrapper fileChannel;
    private final AtomicInteger topicIdSequence = new AtomicInteger();//用于生成递增的 topicId

    public PersistentTopicMap() {
        fileChannel = new FileChannelWrapper(Conf.ssdPath + "topics");
        //从硬盘上恢复 topicName 和分配的 id 对应关系
        if (fileChannel.size() > 0) {
            ByteBuffer head = ByteBuffer.allocate(5);//topicId(1) length(4)
            long readPos = 0;
            while (fileChannel.size() > readPos) {
                head.clear();
                readPos += fileChannel.readAndFlip(head, readPos);
                byte topicId = head.get();
                ByteBuffer topicName = ByteBuffer.allocate(head.getInt());
                readPos += fileChannel.readAndFlip(topicName, readPos);
                topics.computeIfAbsent(StandardCharsets.UTF_8.decode(topicName).toString(), name -> new Topic(name, topicId));
                //更新topicId序列，由于正确性测评断电后不会继续写入，所以这一步没必要
//                if (topicIdSequence.get() < topicId)
//                    topicIdSequence.set(topicId + 1);
            }
        }
    }

    /**
     * 当有新的 topic 时，将topicName 和 topicId 的对应关系写入硬盘
     */
    public Topic getOrCreate(String topicName) {
        return topics.computeIfAbsent(topicName, k -> {
            Topic topic = new Topic(topicName, (byte) topicIdSequence.getAndAdd(1));
            byte[] bytes = topicName.getBytes(StandardCharsets.UTF_8);
            ByteBuffer buffer = ByteBuffer.allocate(bytes.length + 5);
            //topicId(1) length(4) topicName(...)
            buffer.put(topic.id).putInt(bytes.length).put(bytes);
            buffer.flip();
            fileChannel.write(buffer);
            fileChannel.force();
            return topic;
        });
    }

    public Topic get(String topicName) {
        return topics.get(topicName);
    }
}
